package com.gamejelly.broker.client;

import com.alipay.remoting.Connection;
import com.alipay.remoting.config.Configs;
import com.alipay.remoting.exception.RemotingException;
import com.alipay.remoting.rpc.RpcClient;
import com.gamejelly.broker.common.Handler;
import com.gamejelly.broker.message.*;
import com.gamejelly.broker.processor.message.ClientRpcHandler;
import io.netty.util.internal.StringUtil;
import lombok.Data;
import lombok.NonNull;
import org.springframework.core.Ordered;

import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 客户端
 */
@Data
public class BrokerClientImpl implements BrokerClient , Ordered {

    /**
     * 超时时间
     */
    public static final int DEFAULT = 10 * 1000;

    /**
     * bolt-客户端
     */
    private static RpcClient client;

    /**
     * rpc-client-map
     */
    private static Map<String, BrokerClient> clientMap;

    /**
     * 本地地址
     */
    String address;

    /**
     * 中介地址
     */
    @NonNull
    String brokerAddress;

    /**
     * 客户端别名
     */
    @NonNull
    String rpcId;


    /**
     * 回调 rpc消息Id生成器
     */
    private final static AtomicLong MSG_ID_PROVIDER = new AtomicLong();

    /**
     * 回调缓存
     */
    private final static Map<Long, CompletableFuture> futureMap = new ConcurrentHashMap<>();

    /**
     * 同步阻塞
     */
    private final CountDownLatch latch = new CountDownLatch(1);

    private Handler handler;

    /**
     * 数据实体编解码器
     */
    private ClientCodec codec;


    //向broker 注册client
    @Override
    public String registerTo(RegisterMsg msg) throws RemotingException, InterruptedException {
        return (String) client.invokeSync(brokerAddress, msg, DEFAULT);
    }

    //单向发送消息

    private void send(BaseMsg msg) throws RemotingException, InterruptedException {
        client.oneway(brokerAddress, msg);
    }

    public void send(String mainRoute, String subRoute, String endPoint, Object data) throws RemotingException, InterruptedException {
        RpcMessage rpcMessage = genRpcMessage(mainRoute, subRoute, endPoint, data);
        send(rpcMessage);
    }

    @Override
    public <T> T request(String mainRoute, String subRoute, String endPoint, Object data, Class<T> type) throws RemotingException, InterruptedException, ExecutionException, TimeoutException {
        return call(mainRoute, subRoute, endPoint, data, type).get(DEFAULT, TimeUnit.MILLISECONDS);
    }


    @Override
    public <T> CompletableFuture<T> call(String mainRoute, String subRoute, String endPoint, Object data, Class<T> type) throws RemotingException, InterruptedException {
        RpcInfo rpcInfo = new RpcInfo(MSG_ID_PROVIDER.getAndIncrement(), rpcId);
        RpcMessage rpcMessage = genRpcMessage(mainRoute, subRoute, endPoint, data).setRpcInfo(rpcInfo);
        send(rpcMessage);
        CompletableFuture<byte[]> future = new CompletableFuture<>();
        futureMap.put(rpcInfo.getMsgId(), future);
        return future.thenApply(bytes -> codec.decode(bytes, type));
    }

    @Override
    public RpcManager findRpcManager(String name) {
        return clientMap.get(name);
    }

    private RpcMessage genRpcMessage(String mainRoute, String subRoute, String endPoint, Object data) {
        byte[] bytes = codec.encode(data);
        return new RpcMessage().setCmdInfo(CmdInfo.of(mainRoute, subRoute, endPoint)).setData(bytes);
    }


    @Override
    public void handle(RpcMessage rpcMessage) throws InvocationTargetException, IllegalAccessException {
        //处理错误信息
        if (!StringUtil.isNullOrEmpty(rpcMessage.getErr())) {

        } else if (rpcMessage.getRpcInfo() == null || !rpcMessage.getRpcInfo().isCallBack()) {
            //处理request
            RpcResponse rpcResponse = new RpcResponse();
            rpcResponse.setRpcMessage(rpcMessage);
            rpcResponse.setClient(this);
            handler.handle(rpcResponse);
        } else {
            //处理回调
            RpcInfo rpcInfo = rpcMessage.getRpcInfo();
            Optional.of(rpcInfo.getMsgId())
                    .map(futureMap::get)
                    .ifPresent(completableFuture -> completableFuture.complete(rpcMessage.getData()));
        }
    }

    /**
     * 处理连接完成
     */
    public void connected(String hostString) {
        latch.countDown();
        address = hostString;
    }

    @Override
    public void reply(RpcMessage msg, Object o) throws RemotingException, InterruptedException {
        //不是callback
        if (msg.getRpcInfo() == null || msg.getRpcInfo().isCallBack()) {
            return;
        }
        byte[] encode = codec.encode(o);
        msg.getRpcInfo().setCallBack(true);
        msg.setData(encode);
        send(msg);
    }


    /**
     * 初始化一个client
     */
    public void init() throws RemotingException, InterruptedException {
        if (client == null) {
            client = new RpcClient();
            client.registerUserProcessor(new ClientRpcHandler(clientMap));
            client.startup();
        }

        //初始化连接
        Connection connection = client.getConnection(brokerAddress, DEFAULT);
        RegisterMsg msg = new RegisterMsg();
        msg.setRpcId(rpcId);
        if (this.rpcId.equals(registerTo(msg))) {
            connected(connection.getLocalAddress().getAddress().getHostAddress());
        }
        if (codec == null) codec = new DefaultCodec();
    }

    public BrokerClient addRpcHandler(Handler handler) {
        this.handler = handler;
        return this;
    }

    public void addCodec(ClientCodec codec) {
        this.codec = codec;
    }


    private BrokerClientImpl() {
    }

    public static ClientBuilder Builder() {
        return new ClientBuilder();
    }

    @Override
    public int getOrder() {
        return Ordered.LOWEST_PRECEDENCE;
    }


    public static class ClientBuilder {

        String brokerAddr;

        String clientName;

        public ClientBuilder brokerAddr(String addr) {
            this.brokerAddr = addr;
            return this;
        }


        public ClientBuilder clientName(String clientName) {
            this.clientName = clientName;
            return this;
        }

        public BrokerClientImpl Build() {
            System.setProperty(Configs.CONN_MONITOR_SWITCH, "true");
            System.setProperty(Configs.CONN_RECONNECT_SWITCH, "true");
            BrokerClientImpl client = new BrokerClientImpl();
            client.brokerAddress = brokerAddr;
            client.rpcId = clientName;
            if (clientMap == null) {
                clientMap = new ConcurrentHashMap<>();
            }
            clientMap.put(clientName, client);
            return client;
        }

    }

    private BrokerClientImpl aware(Collection<?> collection) {
        for (Object o : collection) {
            if (o instanceof BrokerClientAware)
                ((BrokerClientAware) o).setClient(clientMap);
        }
        return this;
    }
}
